/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.worker;

import cn.ac.iie.di.ban.ycl.wx.nrd.update.commons.RuntimeEnv;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.commons.RuntimeEnvException;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class SenderHandler extends Thread {

    static private final Logger logger = Logger.getLogger(SenderHandler.class.getName());

    static private final HashMap<String, REFieldType> types;

    static {
        types = new HashMap<>();
        try {

            types.put("_ch", REFieldType.Long);
            types.put("_spec", REFieldType.String);
            types.put("ut", REFieldType.Long);

            if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("msg")) {
                types.put("wx_url_fng", REFieldType.String);
                types.put("nrd", REFieldType.Long);
            } else if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("user")) {
                types.put("uid", REFieldType.String);
                types.put("wx_nad", REFieldType.Long);
                types.put("nfans", REFieldType.Long);
                types.put("_id", REFieldType.String);
                types.put("_key", REFieldType.String);
                types.put("nrd", REFieldType.Long);
                types.put("wx_key_fng", REFieldType.String);
            } else if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("wx_ic")) {
                types.put("wx_key_fng", REFieldType.String);
                types.put("wx_ic", REFieldType.Long);
                types.put("_id", REFieldType.String);
                types.put("_key", REFieldType.String);
            }
        } catch (RuntimeEnvException ex) {
            logger.error(ex.getMessage(), ex);
        }

    }

    private final MongoHelper[] mongos;
    private final Random rand;
    private final SenderPool[] pools;
    HashMap<String, Object> hm;
    static private final String[] lookupFields = {"_id", "_key", "_spec", "_ch", "nrd", "ut", "wx_sfrom", "wx_url_fng"};

    public SenderHandler(SenderPool[] senderPools, MongoHelper[] mongos, HashMap<String, Object> hm) {
        this.mongos = mongos;
        this.rand = new Random(System.currentTimeMillis());
        this.pools = senderPools;
//        stopped = false;
//        aliveCount = new AtomicInteger(0);
        this.hm = hm;

    }

    @Override
    public void run() {
        try {
            //get needed cols from message
            HashMap<String, HashMap<String, Object>> valueMap2keyfng = new HashMap<String, HashMap<String, Object>>();
            HashMap<String, HashMap<String, Object>> mqMap = new HashMap<String, HashMap<String, Object>>();

            if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("msg")) {
//                logger.info("hm.size"+hm.size());
                List<HashMap<String, Object>> _idRslt = mongos[1].find("wx_url_fng", hm.keySet(), lookupFields);
//                logger.info("_idRslt" + _idRslt.size());
                String _spec = null;
                Long _ch = 40l;
                String _id = null;
                String _key = null;
                Long nrd = 0l;
                Long ut = (long) System.currentTimeMillis() / 1000;
                for (HashMap<String, Object> returnrec : _idRslt) {
                    HashMap<String, Object> rec1 = new HashMap<String, Object>();
                    HashMap<String, Object> uprec = new HashMap<String, Object>();
                    _id = (String) returnrec.get("_id");
                    _key = (String) returnrec.get("_key");
                    _spec = (String) returnrec.get("_spec");
                    String wx_sfrom = (String) returnrec.get("wx_sfrom");
                    String wx_url_fng = (String) returnrec.get("wx_url_fng");
                    HashMap<String, Object> urlMap = (HashMap<String, Object>) hm.get(wx_url_fng);
                    nrd = (Long) urlMap.get("nrd");
//                    logger.info("_id:" + _id + ",nrd : " + nrd);
                    uprec.put("_spec", _spec);
                    uprec.put("_ch", _ch);
                    uprec.put("ut", ut);
                    uprec.put("wx_url_fng", wx_url_fng);
                    uprec.put("nrd", nrd);
                    mqMap.put(_id, uprec);

                    rec1.put("nrd", nrd);
                    rec1.put("ut", ut);
                    valueMap2keyfng.put(_id, rec1);
                }
            } else if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("user")) {
                String _id = null;
                Long ut = (long) System.currentTimeMillis() / 1000;
                for (Map.Entry<String, Object> entry : hm.entrySet()) {
                    _id = entry.getKey();
                    HashMap<String, Object> value = (HashMap<String, Object>) entry.getValue();
                    if (value.containsKey("uid")) {
//                        logger.info("one message:" + _id + "uid:" + value.get("uid"));
                        value.put("ut", ut);
                        valueMap2keyfng.put(_id, value);
                        mqMap.put(_id, value);
                    } else if (value.containsKey("nfans")) {
//                        logger.info("one message:" + _id + "nfans:" + value.get("nfans"));
                        value.put("ut", ut);
                        valueMap2keyfng.put(_id, value);
                        mqMap.put(_id, value);
                    } else if (value.containsKey("wx_nad")) {
//                        logger.info("one message:" + _id + "wx_nad:" + value.get("wx_nad"));
                        value.put("ut", ut);
                        valueMap2keyfng.put(_id, value);
                        mqMap.put(_id, value);
                    } else if (value.containsKey("nrd")) {
//                        logger.info("one message:" + _id + "nrd:" + value.get("nrd"));
                        value.put("ut", ut);
                        valueMap2keyfng.put(_id, value);
                        mqMap.put(_id, value);
                    }

                }

            } else if (RuntimeEnv.getParamAsString(RuntimeEnv.PROGRAM_TYPE).equals("wx_ic")) {
                String _id = null;
                Long ut = (long) System.currentTimeMillis() / 1000;
                for (Map.Entry<String, Object> entry : hm.entrySet()) {
                    _id = entry.getKey();
                    HashMap<String, Object> value = (HashMap<String, Object>) entry.getValue();
                    if (value.containsKey("wx_ic")) {
                        logger.info("one message:" + _id + "wx_ic:" + value.get("wx_ic"));
                        value.put("ut", ut);
                        valueMap2keyfng.put(_id, value);
                        mqMap.put(_id, value);
                    }

                }

            }

            //update mongo
            for (MongoHelper mongo : mongos) {

                if (valueMap2keyfng.size() > 0) {
                    mongo.update("_id", valueMap2keyfng);

                }

            }
            //send rslts to rmq
            for (SenderPool pool : pools) {
                RESendSession session = pool.BorrowSession();
                for (HashMap<String, Object> rec : mqMap.values()) {
                    for (Map.Entry<String, Object> entry : rec.entrySet()) {
                        if (types.containsKey(entry.getKey()) && entry.getValue() != null) {
                            switch (types.get(entry.getKey())) {
                                case String:
                                    session.setString(entry.getKey(), (String) entry.getValue());
                                    break;
                                case Long:
                                    session.setLong(entry.getKey(), (long) entry.getValue());
                                    break;
                            }
                        }
                    }//end for (Map.Entry<String, Object> entry : rec.entrySet()) {
                    while (true) {
                        try {
                            session.add();
                            break;
                        } catch (Exception e) {
                            logger.warn("cannot add record because of " + e.getMessage() + ". try to flush first.", e);
                            while (true) {
                                try {
                                    session.flush();
                                    break;
                                } catch (Exception ee) {
                                    logger.warn("cannot flush records because of " + ee.getMessage() + ".", ee);
                                }
                            }
                        }
                    }//end while (true) {
                }//end for (HashMap<String, Object> rec : rslt) {
                while (true) {
                    try {
                        session.flush();
                        break;
                    } catch (Exception ee) {
                        logger.warn("cannot flush records because of " + ee.getMessage() + ".", ee);
                    }
                }
                logger.info("send record size: " + mqMap.size());
                if (session != null) {
                    pool.ReturnSession(session);
                }
            }
//            System.exit(0);
        } catch (Exception ex) {
            logger.error("some thing error occur when do delayed init!", ex);
        } finally {
//            aliveCount.getAndDecrement();
        }
    }
    public static void main(String[] args) {
        int j=0;
        for (int i = 0; i < 10; i++) {
            j++;
            System.out.println(j);
        }
    }

}
